package com.dataaccess.dataaccess.support.fluoriteCloud;

import cn.hutool.json.JSONObject;
import com.dataaccess.dataaccess.common.base.BaseService;
import com.dataaccess.dataaccess.web.service.VisualizationVideoAlarmService;
import com.ys.product.ysmq.front.log.Log;
import com.ys.product.ysmq.front.log.LogFactory;
import com.ys.product.ysmq.front.msg.AbstractConsumerMessage.ConsumerCallBack;
import com.ys.product.ysmq.front.msg.StandardConsumerMessage;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.List;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * <p>
 * 异步处理消息demo. 消息管道的回调接口:ConsumerCallBack,是和消息消费在一个线程中的,所以用户拿到消息后,要放到本地队列用单独的线程或线程池(消息量非常大的时候)去处理.
 * 这个demo演示如何异步处理消息,抽象的说明怎么异步处理,用户可借鉴这个,结合自己的场景,实现适合自己的异步处理逻辑.
 * </p>
 * 
 * @author jiangyuechao 2018年8月13日 上午11:09:48
 * @version V1.0
 * @modificationHistory=========================逻辑或功能性重大变更记录
 * @modify by user: {修改人} 2018年8月13日
 * @modify by reason:{方法名}:{原因}
 */
@Service
@Slf4j
public class AsyncConsumerMessage  extends BaseService {

    // 00d317f1511e41f985f2c34b9c057e00
    private static String appKey = "65a6b87b5e994b9dad18316fb9030153";

    // 6e9545953689729c612484ce1e9654a6
    private static String appSecret = "c9d3ec9adf2b314f000f78bb45a1a668";

    // private static Logger logger = LoggerFactory.getLogger(ReadCommitMessagePatternMain.class);

    private static String group = "group1";

    static int printListMaxSize = 100;

    private static Log log = LogFactory.getLog(AsyncConsumerMessage.class);
    @Resource
    VisualizationVideoAlarmService visualizationVideoAlarmService;
    // 异步处理消息的本地队列.
    private static BlockingQueue<List<Object>> msgQueue = new LinkedBlockingQueue<List<Object>>();

//    public static void main(String[] args) {
////        // 启动异步线程
////        AsyncHandleThread asyncHandleThread = new AsyncHandleThread();
////        asyncHandleThread.start();
////        // 启动SDK消费消息
////        init();
////    }

    public void useInit(){
        AsyncHandleThread asyncHandleThread = new AsyncHandleThread();
        asyncHandleThread.start();
        // 启动SDK消费消息
        init();
    }

    public  void init() {
        // 开放平台的url,这是test环境的url https://test.ys7.com:65
        String path = "https://open.ys7.com";
        //
        // 设置你的appkey,appSecret,group
        StandardConsumerMessage consumerMessage = new StandardConsumerMessage(path, appKey, appSecret, group);
        // 设置调用接口的间隔时间,ms
        consumerMessage.setConsumeIntervalTime(3000);
        // 设置消费消息的回调接口,建议把消息放入本地队列,不要在回调函数里做业务处理.
        consumerMessage.setConsumerCallBack(new ConsumerCallBack() {
            @Override
            public void consumerCall(List<Object> msg) {
                try {
                    // 使用队列异步处理消息
                    msgQueue.put(msg);
                } catch (InterruptedException e) {
                    // 用户这里需要处理异常
                    log.error(String.format("放入队列失败,msgs:", msg), e);
                }
            }
        });
        consumerMessage.initClient();
    }

    /**
     * <p>
     * 异步处理消息的线程,消息量非常大的时候,请使用线程池.
     * </p>
     */
     class AsyncHandleThread extends Thread {
        AtomicBoolean isRunning = new AtomicBoolean(true);

        public AsyncHandleThread() {
            super("Async-HandleMsg-Thread");
        }

        @Override
        public void run() {
            while (isRunning.get()) {
                try {
                    List<Object> msgs = msgQueue.take();
                    log.info("异步处理消息,消息数量:", msgs.size());
                    userProcessMsgFunction(msgs);
                } catch (InterruptedException e) {
                    // 用户这里需要处理异常
                    log.error("线程中断异常", e);
                }
            }
        }

        // 关闭线程
        public void shutdown() {
            isRunning.set(false);
        }
    }

    /**
     * <p>
     * 用户自己的处理消息的函数
     * </p>
     * 
     * @author jiangyuechao 2018年8月13日 上午11:21:25
     * @param msgs
     */
     void userProcessMsgFunction(List<Object> msgs) {
        if (msgs.size() > 100) {
            // 消息太多,只打印前100个
            log.info("msgSize:{},msgs:{}", msgs.size(), msgs.subList(0, 100));
        } else {
            log.info("msgSize:{},msgs:{}", msgs.size(), msgs);
        }
        for (Object object : msgs) {
            // class com.alibaba.fastjson.JSONObject
            log.info("my readMessageHandle object:{},class:{}", object, object.getClass());
            com.alibaba.fastjson.JSONObject aliJson = (com.alibaba.fastjson.JSONObject)object;
            JSONObject jsonObject = new JSONObject();
            Set<String> set = aliJson.keySet();
            for(String key : set){
                jsonObject.put(key,aliJson.get(key));
            }
            try {
                visualizationVideoAlarmService.fluoriteCloud(jsonObject);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
}
